/**
 * 非事务生产者发送消息流程
 * 1.构造DefaultMQProducer的实例
 * 2.设置必要属性：如NamesrvAddr、tRetryTimesWhenSendFailed等
 * 3.启动producer
 * 4.构造Message，构造时为其设定Topic、Tag、Body
 * 5.设置message的自定义属性（可用来做SQL过滤）
 * 6.发送消息（4种方式）
 * 7.关闭producer
 * 四种消息发送方法
 * 1.同步：阻塞直到broker状态返回或超时
 * 2.异步：broker状态通过回调返回
 * 3.单向：无需broker返回状态
 * 4.请求响应RPC：阻塞直到消费者返回消息或超时
 */

import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.*;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageBatch;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.apache.rocketmq.remoting.exception.RemotingException;

import java.time.LocalTime;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

public class DemoProducer {
    //同步发送消息，最可靠，性能低，适用于金融等少量高可靠业务场景
    public static void syncProducer() throws Exception {
        //Instantiate with a producer group name.
        DefaultMQProducer producer = new
                DefaultMQProducer("my_group_name1");

        // Specify name server addresses.
        producer.setNamesrvAddr("localhost:9876");
        //同步发送失败的重试次数，默认为2次
        producer.setRetryTimesWhenSendFailed(3);
        //Launch the instance.
        producer.start();
        for (int i = 0; i < 10; i++) {
            //Create a message instance, specifying topic, tag and message body.
            Message msg = new Message("TopicA",   //Topic
                    "TagA||TagD",                  //TAG Filter Data
                    ("Hello RocketMQ Sync" +
                            i + "---" + LocalTime.now()).getBytes(RemotingHelper.DEFAULT_CHARSET) // Message body
            );
            //SQL Filter Data
            msg.putUserProperty("num", "10");
            msg.putUserProperty("color", "blue");
            /*Message构造器重载方法
              Message(String topic, byte[] body)
              Message(String topic, String tags, byte[] body)
              Message(String topic, String tags, String keys, byte[] body)
              Message(String topic, String tags, String keys, int flag, byte[] body, boolean waitStoreMsgOK)
             */

            //延迟消息，设置投递的延迟时间等级，10s后发送此消息
//            msg.setDelayTimeLevel(3);

            //Call send message to deliver message to one of brokers.
            SendResult sendResult = producer.send(msg);

            /*其它重载方法
              send(Message msg, long timeout)
              send(Message msg, MessageQueue mq)
              send(Message msg, MessageQueueSelector selector, Object arg)
              send(Message msg, MessageQueue mq, long timeout)
              send(Message msg, MessageQueueSelector selector, Object arg,long timeout)
             */
            System.out.printf("%s%n", sendResult.getMsgId());
        }
        //Shut down once the producer instance is not longer in use.
        producer.shutdown();
    }


    //异步发送消息，可靠，性能高，适用于大部分业务场景
    public static void asyncProducer() throws Exception {
        //Instantiate with a producer group name.
        DefaultMQProducer producer = new DefaultMQProducer("my_group_name2");
        // Specify name server addresses.
        producer.setNamesrvAddr("localhost:9876");
        //异步发送失败的重试次数，默认为2次
        producer.setRetryTimesWhenSendAsyncFailed(3);
        //默认为关闭，如果开关打开了，会触发发送延迟容错机制来选择发送Queue
        producer.setSendLatencyFaultEnable(true);
        //Launch the instance.
        producer.start();
        producer.setRetryTimesWhenSendAsyncFailed(0);

        int messageCount = 10;
        final CountDownLatch countDownLatch = new CountDownLatch(messageCount);
        for (int i = 0; i < messageCount; i++) {
            try {
                final int index = i;
                Message msg = new Message("TopicB",
                        "TagB",
                        "OrderID188",
                        ("Hello world Async" + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
                producer.send(msg, new SendCallback() {
                    @Override
                    public void onSuccess(SendResult sendResult) {
                        countDownLatch.countDown();
                        System.out.printf("%-10d OK %s %n", index, sendResult.getMsgId());
                    }

                    @Override
                    public void onException(Throwable e) {
                        countDownLatch.countDown();
                        System.out.printf("%-10d Exception %s %n", index, e);
                        e.printStackTrace();
                    }
                });
              /*其它重载方法
                send(Message msg, SendCallback sendCallback, long timeout)
                send(Message msg, MessageQueue mq, SendCallback sendCallback)
                send(Message msg, MessageQueue mq, SendCallback sendCallback, long timeout)
                send(Message msg, MessageQueueSelector selector, Object arg，SendCallback sendCallback)
                send(Message msg, MessageQueueSelector selector, Object arg，SendCallback sendCallback, long timeout)
             */

            } catch (Exception e) {
                e.printStackTrace();
            }
        }
        countDownLatch.await(5, TimeUnit.SECONDS);
        producer.shutdown();
    }

    //单向发送消息，最方便，有消息丢失风险，适用于对消息丢失不敏感业务场景，如日志记录
    public static void onewayProducer() throws Exception {
        //Instantiate with a producer group name.
        DefaultMQProducer producer = new DefaultMQProducer("my_group_name1");
        // Specify name server addresses.
        producer.setNamesrvAddr("localhost:9876");
        //Launch the instance.
        producer.start();

        for (int i = 0; i < 10; i++) {
            //Create a message instance, specifying topic, tag and message body.
            Message msg = new Message("TopicA" /* Topic */,
                    "TagB" /* Tag */,
                    ("Hello RocketMQ Oneway " + i).getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */
            );
            //Call send message to deliver message to one of brokers.
            producer.sendOneway(msg);
            /*其它重载方法
              sendOneway(Message msg, MessageQueue mq)
              sendOneway(Message msg, MessageQueueSelector selector, Object arg)
             */
        }
        //Wait for sending to complete
        Thread.sleep(5000);
        producer.shutdown();
    }

    //向某一特定的队列发送有序消息，单一队列中的消息是有序的（在顺序消费时）
    public static void orderedProducer() throws Exception {
        DefaultMQProducer producer = new DefaultMQProducer("GroupD");
        producer.setNamesrvAddr("localhost:9876");
        producer.start();
        String[] tags = new String[]{"tagA", "tagB", "tagC", "tagD", "tagE"};
        for (int i = 0; i < 50; i++) {
            Message message = new Message("OrderedTopic", tags[i % tags.length], "KEY" + i, ("Ordered Msg:" + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
            SendResult sendResult = producer.send(message, new MessageQueueSelector() {

                //MessageQueueSelector保证发送的消息都存储在同一个MessageQueue即可
                @Override
                public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                    // 消息按照arg被分开放入多个队列, 每个队列中的消息保证按顺序被消费 FIFO，
                    /*int index = (Integer) arg % mqs.size(); //编号对按照消息队列数量取余数来分配
                    System.out.println("QueueSize:" + mqs.size());
                    return mqs.get(index);*/

                    // 消息全部放入同一队列, 全局保持顺序性
                    return mqs.get(0);   //0号队列
                }
            }, i);
            System.out.println(sendResult);
        }
        producer.shutdown();
    }

    //rocketmq-4.7.0推出的“Request-Reply”特性来支持模拟RPC调用：producer发出消息后一直等待consumer回复，回复后才会继续，或超时报异常
    public static void rpcProducer() throws MQClientException, InterruptedException {
        String producerGroup = "rpc_group";
        String topic = "RequestTopic";
        long ttl = 3000;

        DefaultMQProducer producer = new DefaultMQProducer(producerGroup);
        producer.setNamesrvAddr("localhost:9876");
        producer.start();

        try {
            Message msg = new Message(topic,
                    "",
                    "Hello world,RPC Sync".getBytes(RemotingHelper.DEFAULT_CHARSET));

            long begin = System.currentTimeMillis();

            //request代替send发送消息（同步）
            Message retMsg = producer.request(msg, ttl);
            /*其它类似重载方法：
               request( Message msg,  MessageQueue mq,  long timeout)
               request( Message msg,  MessageQueueSelector selector,  Object arg,long timeout)
             */
            long cost = System.currentTimeMillis() - begin;
            System.out.printf("[RPC]request to <%s> cost: %d replyMessage: %s %n", topic, cost, new String(retMsg.getBody()));
        } catch (Exception e) {
            e.printStackTrace();
        }
        producer.shutdown();
    }

    public static void rpcAsyncProducer() throws MQClientException, InterruptedException {
        String producerGroup = "rpc_group";
        String topic = "RequestTopic";
        long ttl = 3000;

        DefaultMQProducer producer = new DefaultMQProducer(producerGroup);
        producer.setNamesrvAddr("localhost:9876");
        producer.start();

        try {
            Message msg = new Message(topic,
                    "",
                    "Hello world,RPC Async!".getBytes(RemotingHelper.DEFAULT_CHARSET));

            long begin = System.currentTimeMillis();
            //request代替send发送消息（异步）
            producer.request(msg, new RequestCallback() {
                @Override
                public void onSuccess(Message message) {
                    long cost = System.currentTimeMillis() - begin;
                    System.out.printf("[RPC]request to <%s> cost: %d replyMessage: %s %n", topic, cost, new String(message.getBody()));
                }

                @Override
                public void onException(Throwable e) {
                    System.err.printf("[RPC]request to <%s> fail.", topic);
                }
            }, ttl);
        /*
         request( Message msg,  MessageQueue mq,  RequestCallback requestCallback, long timeout)
         request( Message msg,  MessageQueueSelector selector,  Object arg,  RequestCallback requestCallback, long timeout)
         */
        } catch (Exception e) {
            e.printStackTrace();
        }
        /* shutdown after your request callback is finished */
//        producer.shutdown();
    }

    //发送集合消息，每条消息最大容量为1M
    public static void batchSyncProducer() throws MQClientException, InterruptedException,RemotingException, MQBrokerException {
        DefaultMQProducer producer = new DefaultMQProducer("BatchProducerGroupName");
        producer.setNamesrvAddr("localhost:9876");
        producer.start();

        //large batch
        String topic = "BatchTest";
        List<Message> messages = new ArrayList<>(100 * 1000);
        for (int i = 0; i < 100 * 1000; i++) {
            messages.add(new Message(topic, "Tag", "OrderID" + i, ("Hello world " + i).getBytes()));
        }

        //split the large batch into small ones:
        ListSplitter splitter = new ListSplitter(messages);
        while (splitter.hasNext()) {
            List<Message> listItem = splitter.next();
            producer.send(listItem);
            /*重载方法：
              send(Collection<Message> msgs)
              send(Collection<Message> msgs,long timeout)
              send(Collection<Message> msgs,MessageQueue messageQueue)
              send(Collection<Message> msgs, MessageQueue messageQueue,long timeout)

              send(Collection<Message> msgs, SendCallback sendCallback)
              send(Collection<Message> msgs, SendCallback sendCallback,long timeout)
              send(Collection<Message> msgs, MessageQueue mq,SendCallback sendCallback)
              send(Collection<Message> msgs, MessageQueue mq,SendCallback sendCallback, long timeout)
             */
        }
    }
}

//将Message拆分成1M字节的一个个子消息
class ListSplitter implements Iterator<List<Message>> {
    private int sizeLimit = 1000 * 1000;    //1M
    private final List<Message> messages;
    private int currIndex;

    public ListSplitter(List<Message> messages) {
        this.messages = messages;
    }

    @Override
    public boolean hasNext() {
        return currIndex < messages.size();
    }

    @Override
    public List<Message> next() {
        int nextIndex = currIndex;
        int totalSize = 0;
        for (; nextIndex < messages.size(); nextIndex++) {
            Message message = messages.get(nextIndex);
            int tmpSize = message.getTopic().length() + message.getBody().length;
            Map<String, String> properties = message.getProperties();
            for (Map.Entry<String, String> entry : properties.entrySet()) {
                tmpSize += entry.getKey().length() + entry.getValue().length();
            }
            tmpSize = tmpSize + 20; //for log overhead
            if (tmpSize > sizeLimit) {
                //it is unexpected that single message exceeds the sizeLimit
                //here just let it go, otherwise it will block the splitting process
                if (nextIndex - currIndex == 0) {
                    //if the next sublist has no element, add this one and then break, otherwise just break
                    nextIndex++;
                }
                break;
            }
            if (tmpSize + totalSize > sizeLimit) {
                break;
            } else {
                totalSize += tmpSize;
            }

        }
        List<Message> subList = messages.subList(currIndex, nextIndex);
        currIndex = nextIndex;
        return subList;
    }

    @Override
    public void remove() {
        throw new UnsupportedOperationException("Not allowed to remove");
    }
}

